/*
 * Copyright 2013-2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.springframework.cloud.gateway.filter;

import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import reactor.ipc.netty.NettyPipeline;
import reactor.ipc.netty.http.client.HttpClient;
import reactor.ipc.netty.http.client.HttpClientRequest;

import java.net.URI;
import java.util.List;
import java.util.Map;

import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.*;

/**
 * @author Spencer Gibb
 */
public class NettyRoutingFilter implements GlobalFilter, Ordered {

	/**基于Netty实现的HttpClient。通过该属性，请求后端的Http服务
	 * 
	 */
	private final HttpClient httpClient;

	public NettyRoutingFilter(HttpClient httpClient) {
		this.httpClient = httpClient;
	}

	@Override
	public int getOrder() {
		// 返回顺序为Integer.MAX_VALUE。在《1GatewayFilter一览》中我们列举了所有GlobalFilter的顺序
		return Ordered.LOWEST_PRECEDENCE;
	}

	@Override
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
	    // 获得 requestUrl
		URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

		// 判断是否能够处理该请求，需要满足两个条件：
		// http://或者https://前缀（Scheme）；
		// 调用ServerWebExchangeUtils#isAlreadyRouted(ServerWebExchange)，判断该请求暂未被其他Routing网关处理
        String scheme = requestUrl.getScheme();
		if (isAlreadyRouted(exchange) || (!scheme.equals("http") && !scheme.equals("https"))) {
			return chain.filter(exchange);
		}

        // 设置该请求已经被处理
		setAlreadyRouted(exchange);

		ServerHttpRequest request = exchange.getRequest();

        // 创建Netty Request Method对象。request#getMethod()返回的不是io.netty.handler.codec.http.HttpMethod，所以需要进行转换
		final HttpMethod method = HttpMethod.valueOf(request.getMethod().toString());

		// 获得 url
		final String url = requestUrl.toString();

		// 创建Netty Request Header对象（io.netty.handler.codec.http.DefaultHttpHeaders），将请求的Header设置给它
		final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
		request.getHeaders().forEach(httpHeaders::set);

		// 请求后端Http服务
		return this.httpClient.request(method, url, req -> {
			// 创建Netty Request对象（reactor.ipc.netty.http.client.HttpClientRequest）
			final HttpClientRequest proxyRequest = req.options(NettyPipeline.SendOptions::flushOnEach)
					// 通过设置clientError=false，下面可以调用Mono#doNext(Consumer)方法，统一订阅处理返回的reactor.ipc.netty.http.client.HttpClientResponse对象
					.failOnClientError(false) // 设置请求失败（后端服务返回响应状体码 >= 400）时，不抛出异常
					// 设置Netty Request对象的Header
					.headers(httpHeaders);

			// 目前是一个BUG ，在2.0.X版本修复。见FormIntegrationTests#formUrlencodedWorks()单元测试的注释说明
			if (MediaType.APPLICATION_FORM_URLENCODED.includes(request.getHeaders().getContentType())) {
				return exchange.getFormData()
						.flatMap(map -> proxyRequest.sendForm(form -> {
							for (Map.Entry<String, List<String>> entry: map.entrySet()) {
								for (String value : entry.getValue()) {
									form.attr(entry.getKey(), value);
								}
							}
						}).then())
						.then(chain.filter(exchange));
			}

			// 请求后端的Http服务
			return proxyRequest.sendHeaders() // 发送请求Header
					.send(request.getBody() // 发送请求Body，其中map(...)的过程为Flux<DataBuffer> => ByteBuffer => Flux<DataBuffer>
							.map(DataBuffer::asByteBuffer) // Flux<DataBuffer> => ByteBuffer
							.map(Unpooled::wrappedBuffer)); // ByteBuffer => Flux<DataBuffer>
		}).doOnNext(res -> {
			// 请求后端Http服务完成，将Netty Response赋值给响应response
			ServerHttpResponse response = exchange.getResponse();
			// 创建org.springframework.http.HttpHeaders对象，将Netty Response Header设置给它，而后设置回给响应response
			HttpHeaders headers = new HttpHeaders();
			res.responseHeaders().forEach(entry -> headers.add(entry.getKey(), entry.getValue()));
			response.getHeaders().putAll(headers);

			// 设置响应response的状态码
			response.setStatusCode(HttpStatus.valueOf(res.status().code()));

			// 设置Netty Response到CLIENT_RESPONSE_ATTR。后续NettyWriteResponseFilter将Netty Response写回给客户端
			// Defer committing the response until all route filters have run
			// Put client response as ServerWebExchange attribute and write response later NettyWriteResponseFilter
			exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
		}).then(chain.filter(exchange)); // 提交过滤器链继续过滤
	}
}
